#include "segment/disk_segment.h"
#include "util/arena/arena_wrapper.h"
#include "util/file_wrapper/file_wrapper.h"

namespace hawking {
namespace indexlib {

DECLARE_string(default_ttl_col_name);
DECLARE_int32(tomb_init_size);

bool DiskSegment::LoadMeta_(const std::string& segment_dir) {
    std::string meta_filename = butil::string_printf("%s/meta", segment_dir.data());
    std::string meta_content = FslibWrapper::ReadFile(meta_filename);

    schema_.reset(new IndexlibIndexConfig);
    bool ret = json2pb::JsonToProtoMessage(meta_content, schema_.get());
    if (!ret) {
        return false;
    }

    document_creator_ = std::make_shared<DocumentCreator>(schema_);
    return true;
}

bool DiskSegment::Load(
    const std::string segment_dir,
    FSFileType data_filetype,
    bool need_increase) {
    segment_dir_ = segment_dir;
    data_filetype_ = data_filetype;
    if (!LoadMeta_(segment_dir)) {
        LOG(ERROR) << "Read and Parse index config fail";
        return false;
    }

    bool ret = LoadInvertedIndex_(segment_dir, data_filetype);
    if (!ret) {
        LOG(ERROR) << "LoadInvertedIndex_ fail";
        return false;
    }

    ret = LoadAttributeIndex_(segment_dir, data_filetype, need_increase);
    if (!ret) {
        LOG(ERROR) << "LoadAttributeIndex_ fail";
        return false;
    }

    ret = InitDeletionIndex_();
    if (!ret) {
        LOG(ERROR) << "InitDeletionIndex_ fail";
        return false;
    }

    ret = LoadPkIndex_(segment_dir);
    if (!ret) {
        LOG(ERROR) << "LoadPkIndex_ fail";
        return false;
    }

    need_increase_ = need_increase;
    if (need_increase_) {
        tomb_stone_ = std::make_unique<TombStone>(FLAGS_tomb_init_size);
    }
    return true;
}

bool DiskSegment::LoadInvertedIndex_(
    const std::string& segment_dir,
    FSFileType data_filetype) {
    std::string inverted_index_dir =
        butil::string_printf("%s/inverted_index", segment_dir.data());
    if (!FslibWrapper::PathExist(inverted_index_dir)) {
        if (!schema_->inverted_configs().empty()) {
            LOG(ERROR) << "LoadInvertedIndex_ Fail. inverted_index dir not found: "
                << inverted_index_dir;
            return false;
        } else {
            return true;
        }
    }

    std::vector<std::string> sub_dirname = FslibWrapper::GetSubDir(inverted_index_dir);    
    if (sub_dirname.size() != schema_->inverted_configs_size()) {
        LOG(ERROR) << "LoadInvertedIndex_ Fail. inverted_index dir has "
            << sub_dirname.size() << " sub dirs, but in inverted_config is "
            << schema_->inverted_configs_size();
        return false;
    }

    std::set<std::string> sub_dir_set, sub_dir_set_cfg;
    sub_dir_set.insert(sub_dirname.begin(), sub_dirname.end());
    for (int32_t idx = 0; idx < schema_->inverted_configs_size(); ++idx) {
        sub_dir_set_cfg.insert(schema_->inverted_configs(idx).field_name());
    }

    if (sub_dir_set != sub_dir_set_cfg) {
        LOG(ERROR) << "LoadInvertedIndex_ Fail. inverted_index dir has "
            << FslibWrapper::Set2String(sub_dir_set, ',')
            << " sub dirs, but in inverted_config is "
            << FslibWrapper::Set2String(sub_dir_set_cfg, ',');
        return false;
    }

    for (const auto& sub_dir : sub_dirname) {
        std::string index_dir =
            butil::string_printf("%s/%s", inverted_index_dir.data(), sub_dir.data());
        int32_t field_id;
        if (!GetFieldIdByName_(sub_dir, &field_id)) {
            LOG(ERROR) << "Unknown sub_dirname " << sub_dir;
            continue;
        }
        std::shared_ptr<InvertedDiskIndexer> disk_indexer =
            std::make_shared<InvertedDiskIndexer>();
        Status ret = disk_indexer->Load(index_dir, data_filetype);
        if (ret != Status::OK) {
            LOG(ERROR) << "LoadInvertedIndex_ Fail when Init with err "
                << static_cast<int32_t>(ret);
            inverted_indexers_.clear();
            return false;
        }

        inverted_indexers_.emplace(field_id, disk_indexer);
    }

    return true;
}

bool DiskSegment::LoadAttributeIndex_(
    const std::string& segment_dir,
    FSFileType data_filetype,
    bool need_increase) {
    // 1. check attribute_index dir exist
    std::string attribute_index_dir =
        butil::string_printf("%s/attribute_index", segment_dir.data());
    if (!FslibWrapper::PathExist(attribute_index_dir)) {
        LOG(ERROR) << "LoadAttributeIndex_ Fail. attribute_index dir not found: "
            << attribute_index_dir;
        return false;
    }

    // 2. check attribute_index subdirs is equal schema_'s fields
    std::vector<std::string> sub_dirname = FslibWrapper::GetSubDir(attribute_index_dir);    
    std::unordered_set<std::string> sub_dir_set, sub_dir_set_cfg;
    sub_dir_set.reserve(sub_dirname.size());
    sub_dir_set.insert(sub_dirname.begin(), sub_dirname.end());
    sub_dir_set_cfg.reserve(schema_->fields_size());
    for (int32_t idx = 0; idx < schema_->fields_size(); ++idx) {
        sub_dir_set_cfg.insert(schema_->fields(idx).field_name());
    }

    if (sub_dir_set != sub_dir_set_cfg) {
        LOG(ERROR) << "LoadAttributeIndex_ Fail. attribute_index dir has "
            << FslibWrapper::Set2String(sub_dir_set, ',')
            << " sub dirs, but in attribute_config is "
            << FslibWrapper::Set2String(sub_dir_set_cfg, ',');
        return false;
    }

    // 3. load every attribute index
    for (const auto& field_config : schema_->fields()) {
        const std::string& field_name = field_config.field_name();
        std::string index_dir =
            butil::string_printf("%s/%s", attribute_index_dir.data(), field_name.data());
        int32_t field_id = -1;
        util::ColumnType field_type;
        int64_t slot_count_for_update = field_config.slot_count_for_update();
        int64_t slot_bytes_for_update = field_config.slot_bytes_for_update();
        GetFieldIdAndTypedByName(field_name, &field_id, &field_type);

        std::shared_ptr<AttributeDiskIndexer> disk_indexer;
        switch (field_type) {
        case util::ColumnType::COLUMN_INT8: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<int8_t>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<int8_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT16: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<int16_t>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<int16_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT32: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<int32_t>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<int32_t>>(
                    field_id, field_type, attr_convertor);
        }
            break;
        case util::ColumnType::COLUMN_INT64: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<int64_t>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<int64_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_FLOAT: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<float>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<float>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_DOUBLE: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<SingleValueAttributeConvertor<double>>(field_type);
            disk_indexer =
                std::make_shared<SingleValueAttributeDiskIndexer<double>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_STRING: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<StringAttributeConvertor>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<char>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT8_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<int8_t>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<int8_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT16_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<int16_t>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<int16_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT32_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<int32_t>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<int32_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_INT64_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<int64_t>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<int64_t>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_FLOAT_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<float>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<float>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_DOUBLE_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiValueAttributeConvertor<double>>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<double>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        case util::ColumnType::COLUMN_STRING_LIST: {
            std::shared_ptr<AttributeConvertor> attr_convertor =
                std::make_shared<MultiStringAttributeConvertor>(field_type);
            disk_indexer =
                std::make_shared<MultiValueAttributeDiskIndexer<std::string>>(
                    field_id, field_type, attr_convertor);
        }
        break;
        default:
            LOG(ERROR) << "LoadAttributeIndex_ Fail. Wrong field_type "
                << static_cast<int32_t>(field_type);
            attribute_indexers_.clear();
            return false;
        }

        Status ret = disk_indexer->Load(
            index_dir, data_filetype, need_increase,
            slot_count_for_update, slot_bytes_for_update);
        if (ret != Status::OK) {
            LOG(ERROR) << "LoadAttributeIndex_ Fail when Init with err "
                << static_cast<int32_t>(ret);
            attribute_indexers_.clear();
            return false;
        }

        attribute_indexers_.emplace(field_id, disk_indexer);
    }

    return true;
}

bool DiskSegment::LoadPkIndex_(const std::string& segment_dir) {
    std::string pk_index_dir =
        butil::string_printf("%s/primary_key", segment_dir.data());
    if (!FslibWrapper::PathExist(pk_index_dir)) {
        LOG(ERROR) << "LoadPkIndex_ Fail. pk_index dir not found: "
            << pk_index_dir;
        return false;
    }

    pk_indexer_ = std::make_shared<PrimaryKeyDiskIndexer>();
    Status ret = pk_indexer_->Load(pk_index_dir, deletion_indexer_);
    if (ret != Status::OK) {
        LOG(ERROR) << "LoadPkIndex_ Fail when Init with err "
            << static_cast<int32_t>(ret);
        return false;
    }

    return true;
}

void DiskSegment::ExcludeTombPks(const HashSet<uint64_t>& tomb_stone_pks) {
    pk_indexer_->ExcludeTombPks(tomb_stone_pks, deletion_indexer_);
}

bool DiskSegment::InitDeletionIndex_() {
    deletion_indexer_ = std::make_shared<DeletionMapIndexer>();
    return deletion_indexer_->Init() == Status::OK;
}

}
}